package cn.hy.dataSync.transport.channel.memory;


import cn.hy.dataSync.config.CoreConstant;
import cn.hy.dataSync.config.TaskConfig;
import cn.hy.dataSync.element.Record;
import cn.hy.dataSync.transport.channel.Channel;
import cn.hy.dataSync.transport.record.TerminateRecord;

import java.util.Collection;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;


public class MemoryChannel extends Channel {

    private int bufferSize;

    private ArrayBlockingQueue<Record> queue;

    private ReentrantLock lock;

    private final Condition notInsufficient;
    private final Condition notEmpty;

    public MemoryChannel(TaskConfig taskConfig) {
        super(taskConfig);
        this.queue = new ArrayBlockingQueue<>(this.getCapacity());
        this.bufferSize = coreConfig.getInteger(CoreConstant.EXCHANGER_BUFFER_SIZE, 1);

        lock = new ReentrantLock();
        notInsufficient = lock.newCondition();
        notEmpty = lock.newCondition();
    }

    @Override
    public void close() {
        super.close();
        try {
            this.queue.put(TerminateRecord.get());
        } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
        }
    }

    @Override
    public void clear() {
        this.queue.clear();
    }

    @Override
    protected void doPush(Record r) {
        try {
            long startTime = System.nanoTime();
            this.queue.put(r);
            waitWriterTime.addAndGet(System.nanoTime() - startTime);
        } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
        }
    }

    @Override
    protected void doPushAll(Collection<Record> rs) {
        try {
            long startTime = System.nanoTime();
            lock.lockInterruptibly();
            while (rs.size() > this.queue.remainingCapacity()) {
                notInsufficient.await(200L, TimeUnit.MILLISECONDS);
            }
            this.queue.addAll(rs);
            waitReaderTime.addAndGet(System.nanoTime() - startTime);
            notEmpty.signalAll();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            lock.unlock();
        }
    }

    @Override
    protected Record doPull() {
        try {
            long startTime = System.nanoTime();
            Record r = this.queue.take();
            waitReaderTime.addAndGet(System.nanoTime() - startTime);
            return r;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(e);
        }
    }

    @Override
    protected void doPullAll(Collection<Record> rs) {
        assert rs != null;
        rs.clear();
        try {
            long startTime = System.nanoTime();
            lock.lockInterruptibly();
            while (this.queue.drainTo(rs, bufferSize) <= 0) {
                notEmpty.await(200L, TimeUnit.MILLISECONDS);
            }
            waitReaderTime.addAndGet(System.nanoTime() - startTime);
            notInsufficient.signalAll();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            lock.unlock();
        }
    }

    @Override
    public int size() {
        return this.queue.size();
    }

    @Override
    public boolean isEmpty() {
        return this.queue.isEmpty();
    }
}
